package cn.edu.thu.laud.lasql.processor;

import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.cql.Attributes;
import org.apache.cassandra.cql.Operation;
import org.apache.cassandra.cql.Term;
import org.apache.cassandra.cql.UpdateStatement;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.utils.ByteBufferUtil;

import cn.edu.thu.laud.db.marshal.LaUDFileType;
import cn.edu.thu.laud.db.marshal.LaUDHblobType;
import cn.edu.thu.laud.db.marshal.LaUDLblobType;
import cn.edu.thu.laud.db.marshal.LaUDSblobType;
import cn.edu.thu.laud.jdbc.util.LaUDJdbcUtils;
import cn.edu.thu.laud.server.thrift.FileInfo;
import cn.edu.thu.laud.thrift.LaUDDataServer;
import cn.edu.thu.laud.thrift.service.LaUDClientState;
import cn.edu.thu.laud.thrift.service.ServiceFileInfo;
import cn.edu.thu.laud.thrift.service.ServiceFileInfo.FileStoreType;

public class LaUDUpdateStatement extends UpdateStatement {

	public LaUDUpdateStatement(String keyspace, String columnFamily,
			String keyName, Map<Term, Operation> columns, List<Term> keys,
			Attributes attrs) {
		super(keyspace, columnFamily, keyName, columns, keys, attrs);
		// TODO Auto-generated constructor stub
	}
	public LaUDUpdateStatement(String keyspace, String columnFamily,
			String keyName, List<Term> columnNames, List<Term> columnValues,
			List<Term> keys, Attributes attrs) {
		super(keyspace, columnFamily, keyName, columnNames, columnValues, keys, attrs);
		// TODO Auto-generated constructor stub
	}
	public Map<String,FileInfo> getFileInfos(String keyspace, ClientState clientState) throws InvalidRequestException{
		CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false);
		if(this.getKeys().size()==0){
			throw new InvalidRequestException("must designed one row");
		}
		String keyString=this.getKeys().get(0).getText();
		FileInfo fileInfo;
		ServiceFileInfo sFileInfo;
		String fileLocation;
		LaUDClientState laUDClientState=(LaUDClientState) clientState;
		List<FileInfo> fileInfos= new ArrayList<FileInfo>();
		Map<String,FileInfo> list=new HashMap<String,FileInfo>();

		for(Term term :this.getColumns().keySet()){
			ColumnDefinition cDefinition=metadata.getColumnDefinition(term.getByteBuffer(metadata.comparator));

			if(cDefinition==null && !(metadata.getDefaultValidator() instanceof LaUDFileType)){
				continue;
			}
			if(cDefinition.getValidator() instanceof LaUDFileType){

				if(this.getKeys().size()>1){
					throw new InvalidRequestException("in file transfer ,only one row can be signed.");
				}
				
					fileLocation="/usr/"+keyspace+"/"+metadata.cfName+"/"+term.getText()+"/"+keyString+"/";
					fileInfo=new FileInfo().setInfo(this.getColumns().get(term).a.getText());
					this.getColumns().put(term,new Operation(new Term("", 7)));
					list.put(term.getText(),fileInfo);
					
					fileInfo.setDirection((byte) LaUDJdbcUtils.WRITE_INTO);
					sFileInfo=new ServiceFileInfo();
					sFileInfo.setLocation(fileLocation);
					sFileInfo.setOverwrite(true);
					
					sFileInfo.keyName=this.keyName;
					//sFileInfo.key=this.getKeys().get(0);
					sFileInfo.keyBuffer=this.getKeys().get(0).getByteBuffer(metadata.getKeyValidator());
					sFileInfo.columnFamily=this.columnFamily;
					//sFileInfo.columnName=term;
					sFileInfo.columnNameBuffer=cDefinition.name;
					sFileInfo.setDirection(LaUDJdbcUtils.WRITE_INTO);
					Attributes attributes=new Attributes();
					attributes.setConsistencyLevel(this.cLevel);
					attributes.setTimestamp(this.timestamp);
					attributes.setTimeToLive(this.timeToLive);
					sFileInfo.attrs=attributes;
					//sFileInfo.columnValue=new Term(fileLocation,7);
					sFileInfo.columnValueBuffer=ByteBufferUtil.bytes(fileLocation);
					
					if(cDefinition.getValidator() instanceof LaUDHblobType){
						sFileInfo.setType(FileStoreType.HDFS);
					}else if(cDefinition.getValidator() instanceof LaUDLblobType){
						sFileInfo.setType(FileStoreType.C_HDFS);
					}else if(cDefinition.getValidator() instanceof  LaUDSblobType){
						sFileInfo.setType(FileStoreType.C_KV);
						sFileInfo.setDirection(LaUDJdbcUtils.WRITE_INTO_CASSANDRA);
						fileInfo.setDirection((byte) LaUDJdbcUtils.WRITE_INTO_CASSANDRA);
					}	
					fileInfo.setCheckcode(laUDClientState.getCheckCode());
					laUDClientState.addFileDataSchedule(fileInfo.getCheckcode(), sFileInfo);
					fileInfos.add(fileInfo);
			}
		}
		if(fileInfos.size()>0)
			LaUDDataServer.authorizedUsers.put(laUDClientState.getSessionId(), laUDClientState);
		return list;
	}


}
